package com.gserver.commons.core.async.impl;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gserver.commons.core.async.AsyncOperation;
import com.gserver.commons.core.async.IAsyncManager;
import com.gserver.commons.core.async.IIoOperation;
import com.gserver.commons.core.async.SyncOperation;
import com.gserver.commons.core.manager.AbstractManager;
import com.gserver.commons.core.server.ICharOperation;
import com.gserver.commons.core.server.IMessageProcessor;

/**
 * 异步操作管理器
 * 
 * 
 * 
 */
public class AsyncManagerImpl extends AbstractManager implements IAsyncManager {

	private static final Logger logger = LoggerFactory.getLogger("async");

	/** 位于主线程的消息处理器,用于当异步操作完成后,在主线程中执行收尾操作 */
	private final IMessageProcessor messageProcessor;
	/** 与玩家角色绑定的线程池,即根据玩家角色的id号{@link PlayerCharacter#getId()},固定的操作总在同一个线程池中运行 */
	private final ExecutorService[] charBindExecutors;
	/** 不与玩家角色绑定的线程池 */
	private final ExecutorService charUnBindExecutor;

	public AsyncManagerImpl(final int charBindExecutorSize,
			final int charUnBindExecutorSize, IMessageProcessor messageProcessor) {
		charBindExecutors = new ExecutorService[charBindExecutorSize];
		for (int i = 0; i < charBindExecutorSize; i++) {
			charBindExecutors[i] = Executors.newSingleThreadExecutor();
		}
		this.messageProcessor = messageProcessor;
		this.charUnBindExecutor = new ThreadPoolExecutor(
				charUnBindExecutorSize, charUnBindExecutorSize * 5, 120L,
				TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
	}


	public AsyncOperation createOperation(IIoOperation operation) {
		if (operation instanceof ICharOperation) {
			int _charId = ((ICharOperation) operation).getCharId();
			int _executorIndex = _charId % this.charBindExecutors.length;
			ExecutorService _asyncExecutor = this.charBindExecutors[_executorIndex];
			return new AsyncOperation(operation, _asyncExecutor,
					messageProcessor);
		} else {
			return new AsyncOperation(operation, charUnBindExecutor,
					messageProcessor);
		}
	}

	public AsyncOperation createOperationAndExecuteAtOnce(IIoOperation operation) {
		AsyncOperation _operation = this.createOperation(operation);
		_operation.execute();
		return _operation;
	}

	public SyncOperation createSyncOperationAndExecuteAtOnce(
			IIoOperation operation) {
		SyncOperation _operation = new SyncOperation(operation);
		_operation.execute();
		return _operation;
	}

	@Override
	public void check() {
	}

	@Override
	public void init() {
	}

	@Override
	public void stop() {
		try {
			for (ExecutorService _executor : this.charBindExecutors) {
				_executor.shutdown();
				// TODO.
				while (!_executor.isTerminated()) {
					try {
						Thread.sleep(1000);
					} catch (Exception e) {
						if (logger.isErrorEnabled()) {
							logger.error(
									"#GS.AsyncManagerImpl.stop.isTerminated", e);
						}
						break;
					}
				}
			}
			this.charUnBindExecutor.shutdown();
			// TODO.
			while (!charUnBindExecutor.isTerminated()) {
				Thread.sleep(1000);
			}
		} catch (Exception e) {
			if (logger.isErrorEnabled()) {
				logger.error("#GS.AsyncManagerImpl.stop", e);
			}
		}
	}
}
